package cn.com.zetatech.maple.rule;

import cn.com.zetatech.maple.entity.Field;
import cn.com.zetatech.maple.entity.HiveWriter;
import cn.com.zetatech.maple.entity.RdbmsWriter;
import cn.com.zetatech.maple.entity.Rule;
import cn.com.zetatech.maple.local.LocalHandler;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;
import org.dom4j.*;

import java.io.File;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * @author cheng.cao@zetatech.com.cn
 */
@Slf4j
public class RuleParser {

    private static final Map<String, Rule> rulesHolder = new ConcurrentHashMap<>();

    public static Rule getFields(String key) {
        return rulesHolder.get(key);
    }

    public static Set<String> getKeys() {
        return rulesHolder.keySet();
    }

    private static String safeGetAttributeVal(Node node, String attr) {
        Node n = node.selectSingleNode("./@" + attr);
        return n == null ? null : ((Attribute) n).getValue();
    }

    private static String safeGetNodeVal(Node node, String xpath) {
        Node n = node.selectSingleNode(xpath);
        return n == null ? null : n.getStringValue();
    }

    private static void parseLocalHandler(Node n, Rule r) {
        Node localHandlerNode = n.selectSingleNode("./local-handler");
        if (localHandlerNode == null) {
            return;
        }
        String handlerClazz = safeGetAttributeVal(localHandlerNode, "class");
        if (handlerClazz == null) {
            return;
        }
        try {
            Class c = Class.forName(handlerClazz);
            Object o = c.newInstance();
            List<Node> props = n.selectNodes("./local-handler/prop");
            if (props != null) {
                for (Node prop : props) {
                    String fieldName = safeGetAttributeVal(prop, "name");
                    String fieldVal = prop.getStringValue();
                    if (fieldName != null) {
                        java.lang.reflect.Field f = c.getDeclaredField(fieldName);
                        f.setAccessible(true);
                        if (fieldVal != null) {
                            f.set(o, fieldVal);
                        }
                    }
                }
            }
            r.setLocalHandler((LocalHandler) o);
        } catch (Exception e) {
            log.error("Fail to parse local-handler => skip", e);
        }
    }

    private static boolean parseWriter(Node n, Rule r) {
        String writer = safeGetNodeVal(n, "./settings/writer/text()");
        if (writer.equalsIgnoreCase("mysql") || writer.equalsIgnoreCase("oracle")) {
            String url = safeGetNodeVal(n, String.format("./settings/%s/url/text()", writer.trim().toLowerCase()));
            String driver = safeGetNodeVal(n, String.format("./settings/%s/driver/text()", writer.trim().toLowerCase()));
            String username = safeGetNodeVal(n, String.format("./settings/%s/username/text()", writer.trim().toLowerCase()));
            String password = safeGetNodeVal(n, String.format("./settings/%s/password/text()", writer.trim().toLowerCase()));
            String table = safeGetNodeVal(n, String.format("./settings/%s/table/text()", writer.trim().toLowerCase()));
            boolean validateResult = validateParam(url, "url") &&
                    validateParam(driver, "driver") &&
                    validateParam(username, "username") &&
                    validateParam(password, "password") &&
                    validateParam(table, "table");
            RdbmsWriter w = new RdbmsWriter();
            w.setUrl(url);
            w.setDriver(driver);
            w.setUsername(username);
            w.setPassword(password);
            w.setTable(table);
            r.setWriter(w);
            return validateResult;
        } else if (writer.equalsIgnoreCase("hive")) {

            String url = safeGetNodeVal(n, String.format("./settings/hive/url/text()", writer.trim().toLowerCase()));
            String driver = safeGetNodeVal(n, String.format("./settings/hive/driver/text()", writer.trim().toLowerCase()));
            String username = safeGetNodeVal(n, String.format("./settings/hive/username/text()", writer.trim().toLowerCase()));
            String password = safeGetNodeVal(n, String.format("./settings/hive/password/text()", writer.trim().toLowerCase()));
            String defaultFs = safeGetNodeVal(n, "./settings/hive/defaultFs/text()");
            String database = safeGetNodeVal(n, "./settings/hive/database/text()");
            String table = safeGetNodeVal(n, String.format("./settings/hive/table/text()", writer.trim().toLowerCase()));
            String fieldsTerminatedBy = safeGetNodeVal(n, String.format("./settings/hive/fieldsTerminatedBy/text()", writer.trim().toLowerCase()));
            String storeBasePath = safeGetNodeVal(n, String.format("./settings/hive/storeBasePath/text()", writer.trim().toLowerCase()));

            List<HiveWriter.Partition> partitions = n.selectNodes("./settings/hive/partitions/partition")
                    .stream()
                    .map(p -> {
                        String name = safeGetNodeVal(p, "./name/text()");
                        String value = safeGetNodeVal(p, "./value/text()");
                        String script = safeGetNodeVal(p, "./script");
                        // value和script两者必有其一
                        if (StringUtils.isNotBlank(name) && StringUtils.isBlank(value) && StringUtils.isBlank(script)) {
                            log.warn("partition需要设置value或者script");
                            return null;
                        }
                        return new HiveWriter.Partition(name, value, script, "groovy");
                    })
                    .filter(p -> p != null)
                    .collect(Collectors.toList());
            Element confElement = (Element) n.selectSingleNode("./settings/hive/conf");
            Map<String, String> conf = new HashMap<>();
            if (confElement != null) {
                confElement.elements().stream().forEach(e -> conf.put(e.getName(), e.getTextTrim()));
            }

            // 分区不参与参数校验，因为可能为非分区表
            boolean validateResult = validateParam(url, "url") &&
                    validateParam(driver, "driver") &&
                    validateParam(username, "username") &&
                    validateParam(password, "password") &&
                    validateParam(defaultFs, "defaultFs") &&
                    validateParam(database, "database") &&
                    validateParam(table, "table") &&
                    validateParam(fieldsTerminatedBy, "fieldsTerminatedBy") &&
                    validateParam(storeBasePath, "storeBasePath");
            HiveWriter w = new HiveWriter();
            w.setUrl(url);
            w.setDriver(driver);
            w.setUsername(username);
            w.setPassword(password);
            w.setDefaultFS(defaultFs);
            w.setDatabase(database);
            w.setTable(table);
            w.setFieldsTerminatedBy(fieldsTerminatedBy);
            w.setStoreBasePath(storeBasePath);
            w.setPartitions(partitions);
            w.setStoreBasePath(storeBasePath);
            w.setConf(conf);
            r.setWriter(w);
            return validateResult;

        } else if (writer.equalsIgnoreCase("hbase")) {
            return false;
        }
        return false;
    }

    private static boolean validateParam(String value, String logName) {
        if (StringUtils.isBlank(value)) {
            log.error("writer下的{}不能未空", logName);
            return false;
        }
        return true;
    }

    public static void load(String configFile) throws Exception {
        String content = FileUtils.readFileToString(new File(configFile), StandardCharsets.UTF_8);
        Document document = DocumentHelper.parseText(content);
        Element root = document.getRootElement();
        Node node = root.selectSingleNode("/rules");
        String namespace = safeGetAttributeVal(node, "namespace");
        String globalEnable = safeGetAttributeVal(node, "enable");
        // 如果不启用的话就直接跳过了
        if ("false".equalsIgnoreCase(globalEnable)) {
            return;
        }
        List<Node> parserList = node.selectNodes("./rule");
        for (Node n : parserList) {
            String itemEnable = safeGetAttributeVal(n, "enable");
            if ("false".equalsIgnoreCase(itemEnable)) {
                continue;
            }
            String id = safeGetAttributeVal(n, "id");
            String parseClazz = safeGetAttributeVal(n, "class");
            String parseType = safeGetAttributeVal(n, "type");
            String separator = safeGetAttributeVal(n, "separator");
            String itemReplace = safeGetAttributeVal(n, "itemReplace");
            String itemSize = safeGetAttributeVal(n, "items");
            String writer = safeGetNodeVal(n, "./settings/writer/text()");

            if (StringUtils.isBlank(writer)) {
                log.error("文件:{}中节点[rule/settings/writer],缺少值,忽略掉该条规则", configFile);
                continue;
            }
            writer = writer.trim().toLowerCase();
            if (
                    !(writer.equalsIgnoreCase("mysql") ||
                            writer.equalsIgnoreCase("oracle") ||
                            writer.equalsIgnoreCase("hive") ||
                            writer.equalsIgnoreCase("hbase"))
            ) {
                log.error("文件:{}中节点[rule/settings/writer],writer={}值错误,忽略掉该条规则", configFile, writer);
                continue;
            }

            if (StringUtils.isBlank(parseClazz) && StringUtils.isBlank(parseType)) {
                log.error("文件:{}中节点[rule],缺少属性type和class(两者必有其一),忽略掉该条规则", configFile);
                continue;
            }
            Rule rule = new Rule();
            if (StringUtils.isNotBlank(parseClazz)) {
                rule.setClazz(parseClazz);
            } else {
                if ("simple".equalsIgnoreCase(parseType)) {
                    if (separator == null || separator.equals("")) {
                        separator = " +";
                    }
                    rule.setSeparator(separator);
                    rule.setItemReplace(itemReplace);

                    if (StringUtils.isNotBlank(itemSize)) {
                        rule.setItemSize(Integer.valueOf(itemSize));
                    }
                    rule.setClazz("cn.com.zetatech.maple.parser.processor.impl.SimpleLineProcessor");
                } else if ("default".equalsIgnoreCase(parseType)) {
                    rule.setClazz("cn.com.zetatech.maple.parser.processor.impl.DeviceLogLineProcessor");
                } else {
                    log.error("文件:{}中节点[rule],不支持的type值，type={},忽略掉该条规则", configFile, parseType);
                    continue;
                }
            }
            if (!parseWriter(n, rule)) {
                continue;
            }

            parseLocalHandler(n, rule);

            {   // 开关: 是否启用本地同类型文件合并，当不设置或者设置为true时，表示启用
                String enableSameTypeLocalMerge = safeGetNodeVal(n, "./settings/enableSameTypeLocalMerge/text()");
                if (StringUtils.isBlank(enableSameTypeLocalMerge) || "true".equalsIgnoreCase(enableSameTypeLocalMerge.trim())) {
                    rule.setEnableSameTypeLocalMerge(true);
                }
            }

            List<Node> mappingFields = n.selectNodes("./mapping/item");
            List<Field> fields = new ArrayList<>();
            for (Node f : mappingFields) {
                Field field = new Field();
                String property = safeGetAttributeVal(f, "property");
                String column = safeGetAttributeVal(f, "column");
                String type = safeGetAttributeVal(f, "type");
                String prefix = safeGetAttributeVal(f, "prefix");
                String scriptEngine = safeGetAttributeVal(f, "scriptEngine");
                String ignore = safeGetAttributeVal(f, "ignore");
                String script = f.getStringValue();
                field.setValue(safeGetAttributeVal(f, "value"));
                field.setScript(script);
                field.setScriptEngine(scriptEngine);
                field.setIgnore("true".equals(ignore));

                if (property == null) {
                    throw new RuntimeException("配置文件错误:property不能未空");
                }
                if (prefix != null) {
                    field.setPrefix(prefix);
                }
                field.setProperty(property);
                // 默认column与property相同
                if (column != null) {
                    field.setColumn(column);
                } else {
                    //复杂类型，字段映射默认规则
                    if (field.getPrefix() != null) {
                        String prefixStr = field.getPrefix().replaceAll("-", "_");
                        // 按照文档上给的演例，如果以name开头的复杂类型字段会将name去掉
                        if (prefixStr.toLowerCase().startsWith("name:")) {
                            prefixStr = prefixStr.substring("name:".length());
                        }
                        //替换：为下划线
                        if (prefixStr.endsWith(":") || prefixStr.endsWith(" ") || prefixStr.endsWith(" ")) {
                            prefixStr = prefixStr.substring(0, prefixStr.length() - 1);
                        }

                        prefixStr = prefixStr.replace(":", "_");
                        if (prefixStr.toLowerCase().endsWith(field.getProperty().toLowerCase())) {
                            field.setColumn(prefixStr);
                        } else {
                            if (field.getProperty().endsWith(".")) {
                                field.setColumn(prefixStr + "_" + field.getProperty().substring(0, field.getProperty().length() - 1));
                            } else {
                                field.setColumn(prefixStr + "_" + field.getProperty());
                            }
                        }
                    } else {
                        field.setColumn(property);
                    }
                }
                if (type != null) {
                    field.setType(type);
                } else {
                    field.setType("string");
                }
                fields.add(field);
            }
            if (fields.isEmpty()) {
                log.error("文件:{}中节点[rule]下没有字段映射,忽略掉该条规则", configFile);
                continue;
            }
            rule.setFields(fields);
            rulesHolder.putIfAbsent(String.format("%s_%s", namespace, id), rule);
        }
    }
}